package com.mls.guardian.remoting.netty;


import com.mls.guardian.config.NettyConfig;
import com.mls.guardian.exception.RemoteSendRequestException;
import com.mls.guardian.exception.RemoteTimeoutException;
import com.mls.guardian.protocol.RemoteCommand;
import com.mls.guardian.remoting.client.Client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by jl on 15/9/9.
 */
public class NettyClient extends  AbstractNetty implements Client {
    private static final Logger log = LoggerFactory.getLogger(NettyClient.class);
    private int port;
    private String host;
    private boolean isKeepLive;

    Bootstrap bootstrap=new Bootstrap();
    private SocketChannel socketChannel;
    private static final EventExecutorGroup group = new DefaultEventExecutorGroup(20);
    private final ConcurrentHashMap<String /* addr */, ChannelWrapper> channelTables =
            new ConcurrentHashMap<String, ChannelWrapper>();

    private final Lock lockChannelTables = new ReentrantLock();
    private static final long LockTimeoutMillis = 3000;
    // 定时器
    private final Timer timer = new Timer("scanResponseService", true);


    public NettyClient(NettyConfig nettyConfig) throws Exception {
        super(nettyConfig);
        this.isKeepLive = false;
        init();
    }



    public NettyClient(String host,int port,NettyConfig nettyConfig) throws Exception {
        super(nettyConfig);
        this.host = host;
        this.port = port;
        this.isKeepLive = true;
        init();
    }

    public void init() throws Exception {
        EventLoopGroup eventLoopGroup=new NioEventLoopGroup();

        bootstrap
            .group(eventLoopGroup)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, isKeepLive)
            .option(ChannelOption.SO_SNDBUF, nettyConfig.getClientSocketSndBufSize())
            .option(ChannelOption.SO_RCVBUF, nettyConfig.getClientSocketRcvBufSize())
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new IdleStateHandler(20, 10, 0));
                    socketChannel.pipeline().addLast(new NettyEncoder());
                    socketChannel.pipeline().addLast(new NettyDecoder());
                    socketChannel.pipeline().addLast(new NettyClientHandler());
                }
            });


        if(isKeepLive){
            try{
                ChannelFuture future =bootstrap.connect(host, port).sync();
                if (future.isSuccess()) {
                    socketChannel = (SocketChannel)future.channel();
                    log.info("long connect server  {}:{} successful",host,port);
                }
            }catch (Exception e){
                e.printStackTrace();
            }

        }


        // 每隔1秒扫描下异步调用超时情况
        this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                    NettyClient.this.scanResponseTable();
                }
                catch (Exception e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);

    }



    private Channel createChannel(String addr) throws Exception{
        ChannelWrapper channelWrapper = channelTables.get(addr);
        if(channelWrapper != null && channelWrapper.isOK()){
            return channelWrapper.getChannel();
        }

        //公共变量要要锁
        if(lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)){
            try {
                boolean createNewConnection = false;
                channelWrapper = this.channelTables.get(addr);
                if (channelWrapper != null) {
                    // channel正常
                    if (channelWrapper.isOK()) {
                        return channelWrapper.getChannel();
                    }
                    // 正在连接，退出锁等待
                    else if (!channelWrapper.getChannelFuture().isDone()) {
                        createNewConnection = false;
                    }
                    // 说明连接不成功
                    else {
                        this.channelTables.remove(addr);
                        createNewConnection = true;
                    }
                }
                // ChannelWrapper不存在
                else {
                    createNewConnection = true;
                }

                if (createNewConnection) {
                    ChannelFuture channelFuture =
                            this.bootstrap.connect(RemoteTools.string2SocketAddress(addr)).sync();
                    log.info("createChannel: start connect [{}]", addr);

                    channelWrapper = new ChannelWrapper(channelFuture);
                    this.channelTables.put(addr, channelWrapper);

                }
            }
            catch (Exception e) {
                log.error("createChannel: create channel error", e);
                throw e;
            }
            finally {
                this.lockChannelTables.unlock();
            }
        }else{
            log.warn("get channelTables lock failed！！！");
        }



        if (channelWrapper != null) {
            ChannelFuture channelFuture = channelWrapper.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(1000)) {
                if (channelWrapper.isOK()) {
                    log.info("createChannel: connect remote host[{}] success, {}", addr,
                            channelFuture.toString());
                    return channelWrapper.getChannel();
                }
                else {
                    log.warn(
                            "createChannel: connect remote host[" + addr + "] failed, "
                                    + channelFuture.toString(), channelFuture.cause());
                }
            }
            else {
                log.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr,
                        1000, channelFuture.toString());
            }
        }

        return null;
    }


    private void closeChannel(final String addr, final Channel channel){
        if (null == channel)
            return;

        final String addrRemote = null == addr ? RemoteTools.parseChannelRemoteAddr(channel) : addr;

        try {
            if (this.lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    final ChannelWrapper prevCW = this.channelTables.get(addrRemote);

                    log.info("closeChannel: begin close the channel[{}] Found: {}", addrRemote,
                            (prevCW != null));

                    if (null == prevCW) {
                        log.info(
                                "closeChannel: the channel[{}] has been removed from the channel table before",
                                addrRemote);
                        removeItemFromTable = false;
                    }
                    else if (prevCW.getChannel() != channel) {
                        log.info(
                                "closeChannel: the channel[{}] has been closed before, and has been created again, nothing to do.",
                                addrRemote);
                        removeItemFromTable = false;
                    }

                    if (removeItemFromTable) {
                        this.channelTables.remove(addrRemote);
                        log.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                    }

                    RemoteTools.closeChannel(channel);
                }
                catch (Exception e) {
                    log.error("closeChannel: close the channel exception", e);
                }
                finally {
                    this.lockChannelTables.unlock();
                }
            }
            else {
                log.warn("closeChannel: try to lock channel table, but timeout, {}ms", LockTimeoutMillis);
            }
        }
        catch (InterruptedException e) {
            log.error("closeChannel exception", e);
        }
    }

    class ChannelWrapper {
        private final ChannelFuture channelFuture;


        public ChannelWrapper(ChannelFuture channelFuture) {
            this.channelFuture = channelFuture;
        }


        public boolean isOK() {
            return (this.channelFuture.channel() != null && this.channelFuture.channel().isActive());
        }


        public boolean isWriteable() {
            return this.channelFuture.channel().isWritable();
        }


        private Channel getChannel() {
            return this.channelFuture.channel();
        }


        public ChannelFuture getChannelFuture() {
            return channelFuture;
        }
    }


    private class NettyClientHandler extends SimpleChannelInboundHandler<RemoteCommand>{

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemoteCommand msg) throws Exception {
            handleReciveMessage(ctx,msg);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.fireExceptionCaught(cause);
        }
    }



    public RemoteCommand sendMessageSync(String addr, final RemoteCommand request, long timeoutMillis)
            throws Exception {
        Channel channel = null;

        try {

            channel = this.createChannel(addr);

            RemoteCommand response = super.sendMessageSync(channel, request, timeoutMillis);

            return response;
        }catch (RemoteSendRequestException e) {
            log.warn("sendMessageSync: send request exception, so close the channel[{}]", addr);
            this.closeChannel(addr, channel);
            throw e;
        }catch (RemoteTimeoutException rte) {
            log.warn("sendMessageSync: wait response timeout exception, the channel[{}]", addr);
            throw rte;
        }

    }



    public RemoteCommand sendMessageSync(final RemoteCommand request, long timeoutMillis)
            throws Exception {
        Channel channel = null;

        try {
            if(socketChannel == null){
                channel = createChannel(host+":"+port);
            }else{
                channel = socketChannel;
            }
            RemoteCommand response = super.sendMessageSync(channel, request, timeoutMillis);

            return response;
        }catch (RemoteSendRequestException e) {
            log.warn("sendMessageSync: send request exception, so close the channel[{}]", channel.remoteAddress());
            throw e;
        }catch (RemoteTimeoutException e) {
            log.warn("sendMessageSync: wait response timeout exception, the channel[{}]", channel.remoteAddress());
            throw e;
        }

    }

    @Override
    public void sendMessageOneWay(Channel channel, RemoteCommand request) throws Exception {

    }

    @Override
    public void sendMessageOneWay(String addr, RemoteCommand request) throws Exception {
        final Channel channel = this.createChannel(addr);
        if (channel != null && channel.isActive()) {
            this.sendMessageOneWayImpl(channel, request);
        }
        else {
            this.closeChannel(addr, channel);
        }

    }

    public void sendMessageOneWayImpl(Channel channel,RemoteCommand request){
        channel.writeAndFlush(request);
    }


    @Override
    public void reconnect() throws Exception {
        if(isKeepLive){
            ChannelFuture future =bootstrap.connect(host, port).sync();
            if (future.isSuccess()) {
                socketChannel = (SocketChannel)future.channel();
                log.info("long connect server  {}:{} successful",host,port);
            }
        }
    }

    public void shutdown(){
        if (this.timer != null) {
            this.timer.cancel();
        }

        if(socketChannel != null){
            log.info("netty client close {}",socketChannel.remoteAddress());
            socketChannel.close();
        }

        for (ChannelWrapper cw : this.channelTables.values()) {
            this.closeChannel(null, cw.getChannel());
        }

        channelTables.clear();
    }

    @Override
    public boolean keepAaliveIsActive() {
        if(socketChannel == null){
            return false;
        }
        return socketChannel.isActive();
    }
}
